package com.bigdata.mapred;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

//Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
class FlowSumMapper extends Mapper<LongWritable, Text, Text, FlowBean> {

	FlowBean flowBean = new FlowBean();

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		String logLine = value.toString();
		String[] words = logLine.split("\\s+");
		flowBean.set(Long.parseLong(words[6]), Long.parseLong(words[7]));
		context.write(new Text(words[0]), flowBean);

	}
}

// Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
class FlowSumReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

	FlowBean v = new FlowBean();

	@Override
	protected void reduce(Text keyIn, Iterable<FlowBean> valIn, Reducer<Text, FlowBean, Text, FlowBean>.Context context)
			throws IOException, InterruptedException {

		long upFlowCount = 0;
		long downFlowCount = 0;

		for (FlowBean bean : valIn) {
			upFlowCount += bean.getUpFlow();
			downFlowCount += bean.getDownFlow();
		}
		v.set(upFlowCount, downFlowCount);
		context.write(keyIn, v);
	}
}

public class FlowSumDriver {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();

		Job job = Job.getInstance(conf);

		job.setJarByClass(FlowSumDriver.class);

		job.setMapperClass(FlowSumMapper.class);
		job.setReducerClass(FlowSumReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(FlowBean.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowBean.class);

		job.setCombinerClass(FlowSumReducer.class);

		// 判断output文件夹是否存在，如果存在则删除
		Path path = new Path("D:\\test\\output");// 取第1个表示输出目录参数（第0个参数是输入目录）
		FileSystem fileSystem = path.getFileSystem(conf);// 根据path找到这个文件
		if (fileSystem.exists(path)) {
			fileSystem.delete(path, true);// true的意思是，就算output有东西，也一带删除
		}

		// 指定本次mr 输入的数据路径 和最终输出结果存放在什么位置
		FileInputFormat.setInputPaths(job, "D:\\test\\input\\wc.txt");
		FileOutputFormat.setOutputPath(job, path);

		boolean flag = job.waitForCompletion(true);
		System.exit(flag ? 0 : 1);

	}

}

class FlowBean implements WritableComparable<FlowBean> {

	private Long upFlow;
	private Long downFlow;
	private Long sumFlow;

	public FlowBean() {
	}

	public FlowBean(Long upFlow, Long downFlow, Long sumFlow) {
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = sumFlow;
	}

	public FlowBean(Long upFlow, Long downFlow) {
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = upFlow + downFlow;
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeLong(upFlow);
		out.writeLong(downFlow);
		out.writeLong(sumFlow);

	}

	@Override
	public void readFields(DataInput in) throws IOException {
		this.upFlow = in.readLong();
		this.downFlow = in.readLong();
		this.sumFlow = in.readLong();
	}

	public void set(Long upFlow, Long downFlow) {
		this.upFlow = upFlow;
		this.downFlow = downFlow;
		this.sumFlow = upFlow + downFlow;
	}

	@Override
	public String toString() {
		return upFlow + "\t" + downFlow + "\t" + sumFlow;
	}

	public Long getUpFlow() {
		return upFlow;
	}

	public void setUpFlow(Long upFlow) {
		this.upFlow = upFlow;
	}

	public Long getDownFlow() {
		return downFlow;
	}

	public void setDownFlow(Long downFlow) {
		this.downFlow = downFlow;
	}

	public Long getSumFlow() {
		return sumFlow;
	}

	public void setSumFlow(Long sumFlow) {
		this.sumFlow = sumFlow;
	}

	@Override
	public int compareTo(FlowBean o) {
		return this.sumFlow > o.sumFlow ? -1 : 1;
	}

}
